package org.code.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.code.util.ByteStringConverter;
import org.code.util.RocketMQUtil;
import org.code.util.ThreadUtil;
import org.junit.Test;

import java.util.List;

/**
 * @author xiao
 * <p>
 * 使用RocketMQUtil发送简单消息
 */
public class RocketMQUtilSendSimple {

    public static final String TOPIC_NAME = "simple-topic";

    @Test
    public void simpleProducer() throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        // 创建默认消息生产者
        DefaultMQProducer producer = RocketMQUtil.getProducer();
        // 创建消息对象
        Message message = new Message(TOPIC_NAME, ByteStringConverter.stringToByteArray("simple message"));
        // 发送消息
        producer.send(message);
        // 关闭消息生产者
        RocketMQUtil.shutdown(producer);
    }

    @Test
    public void simpleConsumer() throws Exception {
        // 创建默认消息消费者
        DefaultMQPushConsumer consumer = RocketMQUtil.getConsumer();
        // 订阅主题
        consumer.subscribe(TOPIC_NAME, "*");
        // 注册监听
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                System.out.println(msg.getMsgId() + ":" +
                        ByteStringConverter.byteArrayToString(msg.getBody()));
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        // 启动消费者
        consumer.start();
        // 阻塞主线程
        ThreadUtil.read();
    }
}
